package com.chenzp.rocketmq.transaction;

import ch.qos.logback.core.util.StringUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

/**
 * @Classname TransactionProducer
 * @Description
 * @Version 1.0.0
 * @Date 2025-06-16 9:57
 * @Created by ncs
 */
@Slf4j
public class TransactionProducer {

    @SneakyThrows
    public static void main(String[] args) {
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("transactionProducer");
        transactionMQProducer.setNamesrvAddr("192.168.125.102:9876");
        ExecutorService executorService = new ThreadPoolExecutor(12, 12, 10, TimeUnit.SECONDS,new LinkedBlockingQueue<>(3),new ThreadFactory(){
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        transactionMQProducer.setExecutorService(executorService);
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                String tags = msg.getTags();
                if(StringUtil.isNullOrEmpty(tags)){
                    return LocalTransactionState.UNKNOW;
                }
                if(tags.contains("tagA")){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else if(tags.contains("tagB")){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String tags = msg.getTags();
                log.info("checkLocalTransaction{}", tags);
                if(StringUtil.isNullOrEmpty(tags)){
                    return LocalTransactionState.UNKNOW;
                }
                if(tags.contains("tagC")){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else if(tags.contains("tagD")){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }else {
                    return LocalTransactionState.UNKNOW;
                }
            }
        });
        transactionMQProducer.start();
        String[] tags = new String[]{"tagA", "tagB", "tagC", "tagD", "tagE"};
        for (int i = 0; i < 5; i++) {
            try {
                Message msg = new Message("TransactionTopic", tags[i % tags.length], "Key" + i, ("事务消息" + i) .getBytes(StandardCharsets.UTF_8));
                TransactionSendResult transactionSendResult = transactionMQProducer.sendMessageInTransaction(msg, null);
                log.info("事务消息发送结果{}", transactionSendResult);
            } catch (Exception e) {
                log.error("事务消息发送失败", e);
            }
        }
    }
}
